package com.atguigu.gmall.realtime.func;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.util.DruidUtil;
import com.atguigu.gmall.realtime.util.GmallConfig;
import com.atguigu.gmall.realtime.util.PhoenixUtil;
import org.apache.commons.lang.StringUtils;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import javax.sql.DataSource;
import java.sql.Connection;
import java.util.Collection;
import java.util.Set;

/**
 * @author caodan
 * @version 1.0
 * @date 2022-09-15 8:47
 */
public class SinkPhoenixFunc extends RichSinkFunction<JSONObject> {

    DataSource dataSource;

    @Override
    public void open(Configuration parameters) throws Exception {
        dataSource = DruidUtil.createDataSource();
    }


    /**
     * 执行插入操作
     *
     * @param jsonObj 插入的jsonObj对象
     * @param context 上下文对象
     * @throws Exception 异常
     */
    @Override
    public void invoke(JSONObject jsonObj, Context context) throws Exception {
        // 获取插入的phoenix表名
        String tableName = jsonObj.getString("sink_table");
        // 获取jsonObj中的键值对
        jsonObj.remove("sink_table");
        Set<String> keySet = jsonObj.keySet();
        Collection<Object> value = jsonObj.values();
        String keys = StringUtils.join(keySet, ",");
        String values = "'" + StringUtils.join(value, "','") + "'";
        String upsertSQL = "upsert into " + GmallConfig.PHOENIX_SCHEMA + "." + tableName + " (" + keys + ") values(" + values + ")";
        Connection conn;
        System.out.println(upsertSQL);
        try {
            conn = dataSource.getConnection();
            PhoenixUtil.executeSQL(upsertSQL,conn);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
